package com.flute.haflute.agent.job;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.topo.ClusterMessageListener;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.jobbox.base.CloudJobConf;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessageFactory;
import com.flute.haflute.net.messaging.MessageSessionPair;
import com.flute.haflute.tools.Utils;

/**
 * 集群监听JOB请求的唯一入口 所有集群任务请求应当由此服务处理
 * 
 */
public class JobListener4Cluster implements ClusterMessageListener {

	private static Logger logger = LoggerFactory
			.getLogger(JobListener4Cluster.class);

	/** @see RootJobScheduler */
	private RootJobScheduler rootScheduler;

	/** jobRequests JOB请求的缓存 */
	private ArrayList<MessageSessionPair> jobRequests;

	public JobListener4Cluster() {
		rootScheduler = new RootJobScheduler();
		jobRequests = new ArrayList<MessageSessionPair>();
		startHandleThread();
	}

	/*
	 * @see com.flute.haflute.agent.topo.ClusterStateListener#handleMessage(com
	 * .flute.haflute.net.messaging.Message,
	 * org.apache.mina.core.session.IoSession)
	 */
	@Override
	public void asyncHandle(Message message, IoSession session) {
		synchronized (jobRequests) {
			jobRequests.add(new MessageSessionPair(message, session));
			jobRequests.notifyAll();
		}
	}

	private void startHandleThread() {
		Thread t = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					if (jobRequests.size() > 0) {
						MessageSessionPair pair;
						synchronized (jobRequests) {
							pair = jobRequests.remove(0);
						}

						allocateJobResource(pair.getMessage(),
								pair.getSession());

					} else {
						try {
							synchronized (jobRequests) {
								if (jobRequests.size() == 0)
									jobRequests.wait();
							}
						} catch (InterruptedException e) {
						}
					}
				}
			}
		});
		t.setName("JobListener4Cluster Dealer");
		t.start();
	}

	private void allocateJobResource(Message message, IoSession session) {
		if (message.getMessageType().equals(Message.MessageType.JOB_NEW.name())) {
			Message response = null;
			try {
				CloudJobConf jobRequest = Utils.deserialize(message
						.getMessageBody());
				// 根据策略初始化任务所需要的资源
				Node master = rootScheduler
						.initialContextAndNotifyNodes(jobRequest);
				response = MessageFactory.getNewJobResponseMessage(true,
						master.getIp());
			} catch (Exception e) {
				response = MessageFactory.getNewJobResponseMessage(false, null);
				logger.error("", e);
			} finally {
				// 通知客户端JOB提交的结果：成功或者失败
				try {
					Utils.sendMessage(session, response);
				} catch (IOException e) {
					logger.error("", e);
				}
			}
		} else if (message.getMessageType().equals(
				Message.MessageType.JOB_ENDED.name())) {
			try {
				CloudJobConf jobRequest = Utils.deserialize(message
						.getMessageBody());
				rootScheduler.recoverNodesInCache(jobRequest);
			} catch (Exception e) {
				logger.error("", e);
			}
		} else if (message.getMessageType().equals(
				Message.MessageType.JOB_SCHEDULED.name())) {
			try {
				CloudJobConf jobRequest = Utils.deserialize(message
						.getMessageBody());
				rootScheduler.clearCache(jobRequest);
			} catch (Exception e) {
				logger.error("", e);
			}

		}
	}
}
